package com.navinfo.platform.trip.analysis.flink.window;

import com.navinfo.platform.trip.analysis.flink.config.SystemConfig;
import com.navinfo.platform.trip.analysis.pojo.PreTripInfo;
import com.navinfo.platform.trip.analysis.pojo.TripJudgeState;
import com.navinfo.platform.trip.analysis.service.ServiceFactory;
import com.navinfo.platform.trip.analysis.service.TripJudgeService;
import com.navinfo.platform.trip.analysis.util.PointComparator;
import com.navinfo.platform.trip.common.enums.ProtocolEnum;
import com.navinfo.platform.trip.common.enums.RunningTypeEnum;
import com.navinfo.platform.trip.common.enums.SystemTypeEnum;
import com.navinfo.platform.trip.common.enums.TripStatusEnum;
import com.navinfo.platform.trip.common.pojo.PointProtocol;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.windowing.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.runtime.operators.windowing.TimestampedValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * 自定义窗口踢除器
 * @author: web
 * @date: 2020-08-24
 **/
public class TripWindowEvictor implements Evictor<PointProtocol, GlobalWindow>{
    private static final Logger logger = LoggerFactory.getLogger(TripWindowEvictor.class);

    /**
     *  运行类型：1流处理，2批处理
     */
    private RunningTypeEnum runningType;

    /**
     * 行程切分服务
     */
    private TripJudgeService tripJudgeService;

    public TripWindowEvictor(ParameterTool jobConfig) {
        //系统类型：1东风，2青汽，3一汽
        SystemTypeEnum systemType = SystemTypeEnum.valueOf(jobConfig.getInt(SystemConfig.SYSTEM_TYPE_KEY, 1));
        runningType = RunningTypeEnum.valueOf(jobConfig.getInt(SystemConfig.RUNNING_TYPE_KEY, 1));
        tripJudgeService = ServiceFactory.createTripJudgeService(systemType, jobConfig);
    }

    @Override
    public void evictBefore(Iterable<TimestampedValue<PointProtocol>> elements, int size, GlobalWindow window, EvictorContext evictorContext) {
        //在此排序没有效果，故不做任何操作
    }
    /**
     * 删除比第一个行程结束点时间早的所有位置点信息，留下比当前行程结束点晚的所有数据
     * @param elements 窗口内的所有数据
     * @param size     窗口大小
     * @param window    窗口对象
     * @param evictorContext  上下文
     */
    @Override
    public void evictAfter(Iterable<TimestampedValue<PointProtocol>> elements, int size, GlobalWindow window, EvictorContext evictorContext) {
        if(null != elements){
            //排序位置点
            List<TimestampedValue<PointProtocol>> points = StreamSupport.stream(elements.spliterator(), true).sorted(new PointComparator<>()).collect(Collectors.toList());

            long now = System.currentTimeMillis();
            long hourLater = now + (60 * 60 * 1000) ;
            long weekAgo = now - (7 * 24 * 60 * 60 * 1000);

            //找出第一个行程结束的时间
            Long tripEndTime = null;
            List<PreTripInfo> tripList = tripJudgeService.judgeTrip(points, null, runningType);
            if(null!=tripList && tripList.size()>0){
                tripEndTime = tripList.get(0).getEndTime();
            }

            if(logger.isDebugEnabled() && points.size()>0){
                logger.debug("[{}]窗口删除前的大小[{}]", points.get(0).getValue().getTid(), size);
            }

            //删除小于当前行程 或 大于1小时之后的数据
            for(Iterator<TimestampedValue<PointProtocol>> it = elements.iterator(); it.hasNext();){
                long gpsTime = it.next().getValue().getGpsTime();
                if((null!=tripEndTime && gpsTime<=tripEndTime) || gpsTime>hourLater || gpsTime<weekAgo){
                    --size;
                    it.remove();
                }
            }

            if(logger.isDebugEnabled() && points.size()>0){
                logger.debug("[{}]窗口删除后的大小[{}]", points.get(0).getValue().getTid(), size);
            }
        }
    }
}
